package org.apache.lorry.demo.producer;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;

public class TopicProducerDemo1 {
    public static void main(String[] args) {
        TopicProducer topicProducer = new TopicProducer("localhost:9092");
        KafkaProducer<String, String> kafkaProducer = topicProducer.getKafkaProducer();
        String value = "{\n" +
                "  \"schema\": {\n" +
                "    \"type\": \"struct\",\n" +
                "    \"fields\": [\n" +
                "      {\n" +
                "        \"type\": \"int64\",\n" +
                "        \"optional\": false,\n" +
                "        \"field\": \"registertime\"\n" +
                "      },\n" +
                "      {\n" +
                "        \"type\": \"string\",\n" +
                "        \"optional\": false,\n" +
                "        \"field\": \"userid\"\n" +
                "      },\n" +
                "      {\n" +
                "        \"type\": \"string\",\n" +
                "        \"optional\": false,\n" +
                "        \"field\": \"regionid\"\n" +
                "      },\n" +
                "      {\n" +
                "        \"type\": \"string\",\n" +
                "        \"optional\": false,\n" +
                "        \"field\": \"gender\"\n" +
                "      }\n" +
                "    ],\n" +
                "    \"optional\": false,\n" +
                "    \"name\": \"ksql.users\"\n" +
                "  },\n" +
                "  \"payload\": {\n" +
                "    \"registertime\": 1493819497170,\n" +
                "    \"userid\": \"User_1\",\n" +
                "    \"regionid\": \"Region_5\",\n" +
                "    \"gender\": \"MALE\"\n" +
                "  }\n" +
                "}\n";
        kafkaProducer.send(new ProducerRecord<>("test-elasticsearch-sink", "key", value));
        kafkaProducer.close();
    }


    @Test
    public void testSendBigData() {
        TopicProducer topicProducer = new TopicProducer("localhost:9092");
        KafkaProducer<String, String> kafkaProducer = topicProducer.getKafkaProducer();
        List<String> bigdataList = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            bigdataList.add("value_" + i);
        }
        kafkaProducer.send(new ProducerRecord<>("test_topic_1", "key_1", JSON.toJSONString(bigdataList)));
        kafkaProducer.close();
    }
}
